/*
 Copyright (C) 2011  Matteo Magnani, Gabriele Nunziante
 
 This program is free software: you can redistribute it and/or modify
 it under the terms of the GNU General Public License as published by
 the Free Software Foundation, either version 3 of the License, or
 (at your option) any later version.
 
 This program is distributed in the hope that it will be useful,
 but WITHOUT ANY WARRANTY; without even the implied warranty of
 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 GNU General Public License for more details.
 
 You should have received a copy of the GNU General Public License
 along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */

/******************************************************************************************************
 * CRS : Client retrieving tweets, conversations and user info and sending them to the Store Server   *
 ******************************************************************************************************
 * 
 ******************************************************************************************************/

import java.io.*;
import java.net.*;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.Statement;
import java.sql.SQLException;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.*;

import twitter4j.*;


public class TweetRetriever 
{
	static int storeServerPort, trendServerPort;
	static String storeServerAddr, trendServerAddr;
	
	
	static BufferedReader in = null;
	static Socket socket = null;
	static Socket storeSocket = null;
	static BufferedReader storeIn = null;
	static PrintStream storeOut = null;
	static UserArray uArray = new UserArray();
	
	/**
	 * @param args[0] CRStoreServer IP Address
	 * @param args[1] CRStoreServer port
	 * @param args[2] TrendServer IP Address
	 * @param args[3] TrendServer port
	 */
	public static void main(String[] args) throws SQLException
	{
		/* DEBUG VARIABLES */
		boolean q = false; /* true to print queries */
		boolean t = false; /* true to print tweets */
		boolean s = false; /* true to print system messages */
		/*                 */
		
		try {
		storeServerAddr = args[0];
		storeServerPort = Integer.parseInt(args[1]);
		
		trendServerAddr = args[2];
		trendServerPort = Integer.parseInt(args[3]);
		}
		catch (Exception e) {
			System.out.println("Usage: java -cp twitter4j-core-2.1.7-SNAPSHOT.jar:. TweetRetriever <storeServerAddr> <storeServerPort> <trendServerAddr> <trendServerPort>");
			System.exit(-1);
		}
				

		while (true) {
			try {
				/* connects to trend server */
				socket = new Socket(trendServerAddr, trendServerPort);
				in = new BufferedReader(new InputStreamReader(socket.getInputStream()));			}
			catch(Exception e) { 
				e.printStackTrace();
				break;
			}
			try {
				/* Reads from CRTrendServer the keyword and the maxID */
				String search = in.readLine();
				if( search != null && search != "" ) {
					String tag = search.split("\t")[0];
					long maxID = Long.parseLong(search.split("\t")[1]);
					
					/* Search for tweets */
					Twitter twitter = new TwitterFactory().getInstance();
					
					List<Tweet> tweets;
					
					try {
						/* Checks if we can do other requests */
						controlContinue(twitter);
						
						Query query = new Query(tag);
						query.setRpp(100);
						query.setMaxId(maxID);
						query.setPage(1);
						//if (q) System.out.println(query.getQuery());
						tweets = twitter.search(query).getTweets();
						if (s) System.out.println("[TweetRetriever] search (tag,maxID): " + tag + "," + maxID + " : " + tweets.size() + " matches");
						if (tweets.size()==0) {
							continue;
						}
						
						DataOutputStream os = new DataOutputStream(socket.getOutputStream());
						os.writeBytes( ""+(tweets.get(tweets.size()-1).getId()-1));
						os.close();
						
						for (Tweet tweet : tweets) {
						    						  
							if (tweet.getToUser()==null) {
								//String user = getUser(twitter, tweet.getFromUser());
								//if (user==null) continue;
								String tw = tweet.getId() + "\t" + tweet.getFromUser() + "\t" + "-1" + "\t" + formatString(tweet.getText()) + "\t" + tweet.getCreatedAt()+ "\t" +
								"-1" + "\t" + "-1" + "\t" + "-1" + "\t" + tweet.getIsoLanguageCode();
								//storeOut.println("user");
								//storeOut.flush();
								//storeOut.println(user);
								//storeOut.flush();
								//if (s) System.out.println("[TweetRetriever] user sent ["+user+"]...");
								//String res = storeIn.readLine();
								//if (res==null || !res.equals("OK")) {
								//	if (s) System.out.println("[ERR] res:"+res);
								//	continue;
								//}		
								//else if (s) System.out.println("[TweetRetriever] OK ");
								connect();
								storeOut.println("tweet");
								storeOut.flush();
								storeOut.println(tw);
								storeOut.flush();
								disconnect();
								if (s) System.out.println("[TweetRetriever] tweet sent ["+tw+"]...");
								//res = storeIn.readLine();
								//if (res==null || !res.equals("OK")) {
								//	if (s) System.out.println("[ERR] res:"+res);
								//	continue;
								//}
								//else if (s) System.out.println("[TweetRetriever] OK");
							}
							else {
								if (s) System.out.println("[TweetRetriever] --CONVERSATION-- " + tweet.getId() + " to " + tweet.getToUser());
								controlContinue(twitter);
								Status st = twitter.showStatus(tweet.getId());
								User u = st.getUser();
								Vector<String> uVec = new Vector<String>();
								Vector<String> tVec = new Vector<String>();
								if (u==null) {
									if (s) System.out.println("[TweetRetriever] user " + tweet.getFromUser() + " not found");
									continue; // skip conversation and go to next tweet from original results
								}
								String tw = st.getId() + "\t" + u.getScreenName() + "\t" + u.getId() + "\t" + st.getText() + "\t" + st.getCreatedAt() + "\t" + 
									st.getInReplyToStatusId() + "\t" + "$$$ORIGIN$$$" + "\t" + st.getRetweetCount() + "\t" + u.getLang();
								tVec.add(tw);
								String user = u.getId() + "\t" + u.getName() + "\t" + u.getScreenName() + "\t" + u.getProfileImageURL() + "\t" + u.getFollowersCount();
								uVec.add(user);
								long prev = st.getInReplyToStatusId();
								while (prev!=-1) {
									if (s) System.out.println("[TweetRetriever] size: " + uVec.size() + " prev: " + prev);
									controlContinue(twitter);
									st = twitter.showStatus(prev);
									u = st.getUser();
									if (u==null)
										break;
									tw = st.getId() + "\t" + u.getScreenName() + "\t" + u.getId() + "\t" + formatString(st.getText()) + "\t" + st.getCreatedAt() + "\t" + 
									st.getInReplyToStatusId() + "\t" + "$$$ORIGIN$$$" + "\t" + st.getRetweetCount() + "\t" + u.getLang();
									tVec.add(tw);
									user = u.getId() + "\t" + u.getName() + "\t" + u.getScreenName() + "\t" + u.getProfileImageURL() + "\t" + u.getFollowersCount();
									uVec.add(user);
									prev = st.getInReplyToStatusId();
								}
								for (int i=uVec.size()-1; i>=0; i--) {
									connect();
									storeOut.println("user");
									storeOut.flush();
									storeOut.println(uVec.elementAt(i));
									storeOut.flush();
									disconnect();
									if (s) System.out.println("[TweetRetriever] user sent ["+uVec.elementAt(i)+"]...");
									//String res = storeIn.readLine();
									//if (res==null || !res.equals("OK")) {
									//	if (s) System.out.println("[ERR] res:"+res);
									//	continue;
									//}								
									//else if (s) System.out.println("[TweetRetriever] OK ");
								}
								for (int i=tVec.size()-1; i>=0; i--) {
									connect();									
									storeOut.println("tweet");
									storeOut.flush();
									storeOut.println(tVec.elementAt(i).replace("$$$ORIGIN$$$",st.getId()+""));	
									storeOut.flush();
									disconnect();
									if (s) System.out.println("[TweetRetriever] tweet sent ["+tVec.elementAt(i).replace("$$$ORIGIN$$$",st.getId()+"")+"]...");
									//String res = storeIn.readLine();
									//if (res==null || !res.equals("OK")) {
									//	if (s) System.out.println("[ERR] res:"+res);
									//	break;
									//}							
									//else if (s) System.out.println("[TweetRetriever] OK ");						
								}
							}							
						}
					} catch (Exception e1) {
						if (s) System.out.println("[TweetRetriever] Could not get tweet info:");
						e1.printStackTrace();
					}
				}
			}
			catch (Exception ex) {
				if (s) System.out.println("[TweetRetriever] Exited from insertions:");
				ex.printStackTrace();
				continue;
			}
			try {
				in.close();
				socket.close();
			}
			catch(Exception e) { 
				e.printStackTrace();
				break;
			}
		}
		
		/* cleaning resources */
		try {
			in.close();
			socket.close();
		} catch (IOException e) {}

		System.out.println("+ TweetRetriever killed");
	}
	
	/* connects to store server */
	public static void connect() throws IOException {
		storeSocket = new Socket(storeServerAddr, storeServerPort);
		storeIn = new BufferedReader(new InputStreamReader(storeSocket.getInputStream()));
		storeOut = new PrintStream(storeSocket.getOutputStream(),true);
	}
		
	public static void disconnect() throws IOException {
		storeIn.close();
		storeOut.close();
		storeSocket.close();
	}
	
	
	/* Checks if we have requests left. If not, sleeps.  */
	private static void controlContinue(Twitter twitter) throws Exception {
		/* Number of requests to save */
		int remaining_hits = 2;
		
		/* Checks remaining requests for this IP address */
		RateLimitStatus status = twitter.getRateLimitStatus();
		
		if( status.getRemainingHits() < remaining_hits ) {
			System.out.println("Few requests left: " + status.getRemainingHits()
					+ " - sleeping for: " + (status.getSecondsUntilReset()+5) +" sec ("+(status.getSecondsUntilReset()/60)+" min) from "+now());

			
			// Closes connection
			try {
				storeIn.close();
				storeOut.close();
				storeSocket.close();
			}
			catch(Exception e) { 
				e.printStackTrace();
			}
			
			//Sleeps
			Thread.sleep( (twitter.getRateLimitStatus().getSecondsUntilReset()*1000)+(5*1000) );

			// Re-open connection
			
			storeSocket = new Socket(storeServerAddr, storeServerPort);
			storeIn = new BufferedReader(new InputStreamReader(storeSocket.getInputStream()));
			storeOut = new PrintStream(storeSocket.getOutputStream(),true);
		}
		
	}	
	
	private static String getUser(Twitter twitter, String userName) {
		if (uArray.contains(userName)) {
			return uArray.get(userName);
		}
		User u = null;
		try {
			controlContinue(twitter);
			u = twitter.showUser(userName);
		} catch (Exception e) {
		}
		if (u == null)
			return null;
		else {
			String userStr = u.getId() + "\t" + u.getName() + "\t" + u.getScreenName() + "\t" + u.getProfileImageURL() + "\t" + u.getFollowersCount();
			uArray.add(userStr);
			return userStr;
		}
	}
	
	private static String formatDate(String date)
	{
		String date_format;
		StringTokenizer st = new StringTokenizer(date, " ");
		String[] date_token = new String[st.countTokens()];

		int i = 0;
		while(st.hasMoreTokens())
		{
		  String s = st.nextToken();
		  date_token[i] = s;
		  i++;
		}
		date_token[1] = date_token[1].replace("Jan", "01");
		date_token[1] = date_token[1].replace("Feb", "02");
		date_token[1] = date_token[1].replace("Mar", "03");
		date_token[1] = date_token[1].replace("Apr", "04");
		date_token[1] = date_token[1].replace("May", "05");
		date_token[1] = date_token[1].replace("Jun", "06");
		date_token[1] = date_token[1].replace("Jul", "07");
		date_token[1] = date_token[1].replace("Aug", "08");
		date_token[1] = date_token[1].replace("Sep", "09");
		date_token[1] = date_token[1].replace("Oct", "10");
		date_token[1] = date_token[1].replace("Nov", "11");
		date_token[1] = date_token[1].replace("Dec", "12");
	
		date_format = date_token[5]+"-"+date_token[1]+"-"+date_token[2]+" "+date_token[3];
		
		return date_format;
	}
	
	
	public static String now() 
	{
		String DATE_FORMAT_NOW = "yyyy-MM-dd HH:mm:ss";
		Calendar cal = Calendar.getInstance();
		SimpleDateFormat sdf = new SimpleDateFormat(DATE_FORMAT_NOW);
		return sdf.format(cal.getTime());
	}
	
	public static String formatString(String txt) {
		return txt.replace("\\"," ").replace("\n"," ").replace("\t"," ").replace("'"," ").replace("\""," ").trim();
	}
}

class UserArray {
	private int maxSize = 10000;
	private Vector<String> uVec;
	private Map<String,Integer> uMap;
	int current = 0;
	
	UserArray() {
		uVec = new Vector<String>();
		uMap = new HashMap<String,Integer>();	}
	
	UserArray(int maxSize) {
		this.maxSize = maxSize;
		uVec = new Vector<String>();
		uMap = new HashMap<String,Integer>();
	}
	
	boolean contains(String user) {
		boolean res = uMap.containsKey(user);
		if (res) System.out.println("[UserArray] MATCH!");
		return res;
	}
	
	// call contains first FIXME
	String get(String user) {
		return uVec.elementAt(uMap.get(user));
	}
	
	void add(String user) {
		if (uVec.size()>=maxSize) {
			String u = uVec.elementAt(current).split("\t")[0];
			uMap.remove(u);
			uMap.put(user.split("\t")[0],current);
			uVec.set(current,user);
			
			current++;
			if (current==maxSize) current = 0;
		}
		else { 
			uVec.add(user);
			uMap.put(user.split("\t")[0],uVec.size()-1);
		}
	}
}
